package msgQueue

import (
	"bytes"
	"encoding/binary"
	"fmt"
	. "gitee.com/kingzyt/common/bufferPool"
	. "gitee.com/kingzyt/common/bytes"
	. "gitee.com/kingzyt/common/countPool"
	"gitee.com/kingzyt/common/log"
	"gitee.com/kingzyt/common/streamWorker"
	. "gitee.com/kingzyt/common/util"
	"io"
	"sync"
)

const default_send_msg_chan_size = 1000
const default_data_temp_buf_size = 1024 // 会根据需要扩展，不需要用户定义
const default_receive_msg_chan_size = 1000
const default_max_receive_msg_size = 2048

const (
	status_closed int32 = iota
	status_running
	status_closing
)

type MsgQueue struct {
	name              string
	sw                *streamWorker.StreamWorker
	procGoLimiter     *CountPool
	sendMsgChan       chan *Element
	sendMsgChanSize   int
	status            AtomicInt32
	changeStatusMutex sync.Mutex

	// dataTempBuf、dataSplitRltTemp 在read中依次被顺序使用，不会有两个goroutine同时操作
	dataTempBuf       []byte
	dataSplitRltTemp  [][]byte
	dataSplitRltTemp2 [][]byte

	customerMsgProc func(marks []uint32, msg []byte)
	closeSign       sync.WaitGroup

	onStoppedAfter func()

	readCount  AtomicInt32
	writeCount AtomicInt32
	readBytes  AtomicInt64
	writeBytes AtomicInt64
}

func (self *MsgQueue) ResetRecord() {
	self.readCount.Set(0)
	self.writeCount.Set(0)
	self.readBytes.Set(0)
	self.writeBytes.Set(0)
}
func (self *MsgQueue) GetRecord() (rc int32, wc int32, rb int64, wb int64) {
	return self.readCount.Get(), self.writeCount.Get(), self.readBytes.Get(), self.writeBytes.Get()
}

func (self *MsgQueue) GetName() string {
	return self.name
}

func (self *MsgQueue) Closed() bool {
	return self.status.Get() == status_closed
}

type NewParams struct {
	SendMsgChanSize    int
	ReceiveMsgChanSize int
	/* 这里仅仅是一个hint，实际若读取的msg超过这个尺寸，会分多次读取。但设定越接近实际大小，越有效率，
	但因为涉及到空间分配，若实际msg非常大，也可以考虑设定小一点，分多次读取，节省缓存空间
	一般就取在所有msg中,在整体流量中占比重最大头的一组msg来做样本,
	取一个这些大头msg中最大的值即可,这样即满足大多数情况下的效率,又保证不会闲置过多内存*/
	MaxReceiveMsgSize int
}

func New(name string, params NewParams) *MsgQueue {
	if params.SendMsgChanSize <= 0 {
		params.SendMsgChanSize = default_send_msg_chan_size
	}
	if params.ReceiveMsgChanSize <= 0 {
		params.ReceiveMsgChanSize = default_receive_msg_chan_size
	}
	if params.MaxReceiveMsgSize <= 0 {
		params.MaxReceiveMsgSize = default_max_receive_msg_size
	}

	rlt := &MsgQueue{
		name:            name,
		procGoLimiter:   GlobalGoroutinePool,
		sendMsgChanSize: params.SendMsgChanSize,
		dataTempBuf:     make([]byte, default_data_temp_buf_size),
	}

	rlt.sw = streamWorker.New(name+"_sw", params.ReceiveMsgChanSize, params.MaxReceiveMsgSize, streamWorker.Close_type_finishall)
	rlt.sw.SetStopParams(streamWorker.StopParams{
		OnStoppedAfter: func() {
			//当接收停止时，closeSign+1
			rlt.closeSign.Done()
			log.Info(rlt.name, log.Trace, "read stopped")
		},
	})

	return rlt
}

type StopParams struct {
	OnStoppedAfter func()
}

func (self *MsgQueue) SetStopParams(params StopParams) {
	self.onStoppedAfter = params.OnStoppedAfter
}

type OpenParams struct {
	OnOpenOK func()

	Reader         io.Reader
	ReadEOFProc    func()
	ReadErrProc    func(err error)
	ReadBeforeProc func()
	ReadAfterProc  func()

	Writer          io.Writer
	WriteErrProc    func(err error)
	WriteBeforeProc func()
	WriteAfterProc  func()

	MsgProc func(marks []uint32, msg []byte)
}

func (self *MsgQueue) Open(openParams OpenParams) bool {
	if !self.changeStatus(status_closed, status_running,
		func() {
			self.sendMsgChan = make(chan *Element, self.sendMsgChanSize)
			self.clearDataTempBuf()
			self.closeSign.Add(2) // 1个是接收，1个是发送
		}, nil) {
		return false
	}

	if openParams.MsgProc == nil {
		panic(fmt.Errorf("%s: MsgProc is nil", self.name))
	}
	self.customerMsgProc = openParams.MsgProc

	writer := openParams.Writer
	writeErrPorc := openParams.WriteErrProc
	writeBeforeProc := openParams.WriteBeforeProc
	writeAfterProc := openParams.WriteAfterProc

	streamWorkerStartParams := streamWorker.StartParams{
		OnStartOK: openParams.OnOpenOK,
		Reader:    openParams.Reader,
		DataProc: func(data []byte) {
			self.procReadData(data)
		},
		ReadEOFProc: func() {
			if openParams.ReadEOFProc != nil {
				openParams.ReadEOFProc()
			}
			self.Close()
		},
		ReadErrProc: func(err error) {
			if openParams.ReadErrProc != nil {
				openParams.ReadErrProc(err)
			}
			self.Close()
		},
		ReadBeforeProc: openParams.ReadBeforeProc,
		ReadAfterProc:  openParams.ReadAfterProc,
	}

	if !self.sw.Start(streamWorkerStartParams) {
		panic(fmt.Errorf("%s: StreamWorker start failed", self.name))
	}

	/* 这里的write没有与streamWorker合并，是因为
	1. msg空间不由msgQueue管理，不能简单地将msg加入chan，
		而且需要pack msg，所以采用了pkgContainer作为chan的元素，但这样就不能与通用的streamWorker合并
	2. write的msg空间是不定的，streamWorker中的空间原则上可以任意，因为read可以对一个大数据做多次read，
		但多次写的话，要么直接用msg来写，但这样就必须要求msg在写之前不能释放，这样就极大限定了外部的使用方式
		要么缓存下来，可以将PkgContainer嵌入write流程，但这样的话，因为这里的实现中，send的就必须是Element，
		需要在外部Pull buf，然后在外部Push buf，实现细节暴露太多
	总之，实现上具体为msgQueue特化的细节太多，不适合作为通用结构，于是无法和streamWorker合并
	*/
	go func() {
		defer func() {
			// 释放所有没有发送出去的pkg的空间
			for pkgContainer := range self.sendMsgChan {
				self.recoverPkg(pkgContainer)
			}

			self.closeSign.Done()
			log.Info(self.name, log.Trace, "write stopped")
		}()

		for pkgContainer := range self.sendMsgChan {
			/* 若close，则直接关闭写，没有处理的直接丢弃，
			不然可能发生，因为本端的读关闭，造成对端写入阻塞，
			进而阻塞读取，最终影响到本端的写入，造成阻塞*/
			if self.status.Get() == status_closing {
				break
			}

			if writeBeforeProc != nil {
				writeBeforeProc() // e.g. setdeadline
			}

			pkg := pkgContainer.Value
			pkgLen := len(pkg)

			for {
				n, err := writer.Write(pkg)
				if err != nil {
					self.recoverPkg(pkgContainer)
					/* 若发生在调用Close之后，发送剩余数据的时候出现错误的话，
					因为状态已经是closing了，再次调用Close会直接退出，没有影响*/
					self.Close()

					if writeErrPorc != nil {
						writeErrPorc(err)
					}
					return
				}

				self.writeCount.Add(1)
				self.writeBytes.Add(int64(n))

				if n < pkgLen {
					pkg = pkg[n:]
					pkgLen -= n
				} else {
					break
				}
			}

			self.recoverPkg(pkgContainer)

			if writeAfterProc != nil {
				writeAfterProc()
			}
		}
	}()

	return true
}

// 这里保证在status_running的时候，无论调用多少次Close，只有一次会真正执行
/* Close之后，立即发生接收和发送的同时启动关闭，
不再接收Close之后的数据，Close当时没有发送成功的数据被直接丢弃，
但处理掉已经接收的数据*/
func (self *MsgQueue) Close() bool {
	if !self.changeStatus(status_running, status_closing, func() {
		close(self.sendMsgChan)
	}, nil) {
		return false
	}

	go func() {
		self.sw.Stop()

		//当发送和接收都停止了之后，再将状态置为closed
		self.closeSign.Wait()

		if self.changeStatus(status_closing, status_closed, nil, nil) {
			if self.onStoppedAfter != nil {
				self.onStoppedAfter()
			}
		} else {
			log.Error(self.name, "status err, need status_closing, now:%d", self.status.Get())
		}
	}()

	return true
}

/* msg的空间由外部负责释放，这里不能管理msg，
而且pack也需要重新申请空间，所以自然地就与原来的msg空间无关了，
外部调用完SendMsg就可以释放msg空间了

同时，因为sendMsgChan在这里是发送端，而chan的关闭最好是发送端来执行，
但我们无法依赖外部的调用来关闭chan，
因为这里可能在判定running之后马上因为调用Close变成closing，通道关闭，
就使用recover捕获平息，返回false

不能使用锁，因为锁就强制所有的发送必须同步进行，而发送间实际是没有关系的，而且chan本身已经保证同步，没有必要再加一个降低效率的锁
不能在这里使用读写锁，因为若sendMsgChan已经满了，那么SendMsg就会大量阻塞，等待write处理，而当writeErr发生时，write流程中会调用close去关闭流程，
这里会调用写锁来改变状态，而此时有大量读锁还未释放，于是write流程在等待sendMsg结束，而sendMsg在等待write结束，造成死锁

忽略向已经关闭的sendMsgChan写msg引发的panic是各种方案中最满足利大弊小的，
这样的panic数量只会在close的那一瞬间可能出现，以后sendMsg再判定的时候，状态已经closing/closed了就没有问题了
*/
//cKey是在LR/GR上的占位号，位置总量在10w，限制连接上限；cId是当前运行时的线路标记，区分同一个占位上的新旧线路
func (self *MsgQueue) SendMsg(marks []uint32, msg []byte) (rlt bool) {
	if self.status.Get() != status_running {
		return
	}

	pkgContainer := self.packMsg(marks, msg)

	defer func() {
		if r := recover(); r != nil {
			//log.Info(self.name, log.Trace, "sendmsg after chan closed")
			//因为msg已经发送失败了，直接回收pkg的空间
			self.recoverPkg(pkgContainer)
		}
	}()

	self.sendMsgChan <- pkgContainer

	rlt = true
	return
}

func (self *MsgQueue) GetStatus() int32 {
	return self.status.Get()
}

var check_code_table = []uint32{7, 4, 2, 9, 1, 6, 7, 8, 10, 6, 3}

const (
	msg_delim_str = "\xff\r\n\r\n\r\xff"
	len_msg_delim = len(msg_delim_str)
	msg_len_len   = 4
	// 若要添加压缩，则在包尾中添加zip_flag_byte，可以是pkg = msg | check_code(4) | marks_count(1) | zip_flag_byte(1) | first_byte(1) | msg_len(4) | msg_delim(5)
	// zip_flag_byte_len   = 1
	first_byte_len  = 1
	marks_count_len = 1
	check_code_len  = 4

	pkg_tail_len = check_code_len + marks_count_len + first_byte_len + msg_len_len + len_msg_delim
)

var msg_delim []byte = []byte(msg_delim_str)

// cid & ckey is not in the check
func (self *MsgQueue) calCheckCode(marksCount byte, firstByte byte, msgAndMarksLen int) (checkCode uint32) {
	checkCode = uint32(marksCount)*check_code_table[0] + uint32(firstByte)*check_code_table[1] + uint32(msgAndMarksLen)*check_code_table[2]
	for i := 0; i < len_msg_delim; i++ {
		checkCode += uint32(msg_delim[i]) * check_code_table[3+i]
	}

	return
}

//thread safe
// pkg = msg | check_code(4) | marks_count(1) | first_byte(1) | msg_len(4) | msg_delim(5)
func (self *MsgQueue) packMsg(marks []uint32, msg []byte) (pkgContainer *Element) {
	msgLen := len(msg)
	marksCount := len(marks)
	if marksCount > 255 {
		panic(fmt.Errorf("%s, msg marks count should be not over 255", self.name))
	}
	msgAndMarksLen := msgLen + marksCount*4
	pkgSize := msgAndMarksLen + pkg_tail_len
	pkgContainer = GlobalBufPull(pkgSize)
	pkg := pkgContainer.Value

	var msgFirstByte byte
	if msgLen > 0 {
		msgFirstByte = msg[0]
	}

	checkCode := self.calCheckCode(byte(marksCount), msgFirstByte, msgAndMarksLen)

	copy(pkg[:msgLen], msg)

	lb := msgLen
	if marksCount > 0 {
		for i := 0; i < marksCount; i++ {
			le := lb + 4
			binary.BigEndian.PutUint32(pkg[lb:le], marks[i])
			lb = le
		}
	}

	l0 := lb + check_code_len
	l1 := l0 + marks_count_len
	l2 := l1 + first_byte_len
	l3 := l2 + msg_len_len

	binary.BigEndian.PutUint32(pkg[lb:l0], checkCode)
	pkg[l0] = byte(marksCount)
	pkg[l1] = msgFirstByte
	binary.BigEndian.PutUint32(pkg[l2:l3], uint32(msgAndMarksLen))
	copy(pkg[l3:], msg_delim)

	return
}
func (self *MsgQueue) recoverPkg(pkgContainer *Element) {
	GlobalBufPush(pkgContainer)
}

func (self *MsgQueue) pushDataTempBuf(frag []byte) {
	self.dataTempBuf = append(self.dataTempBuf, frag...)
}
func (self *MsgQueue) clearDataTempBuf() {
	if len(self.dataTempBuf) > 0 {
		self.dataTempBuf = self.dataTempBuf[0:0]
	}
}
func (self *MsgQueue) dataTempBufEmpty() bool {
	return len(self.dataTempBuf) == 0
}

/*
要注意rlttemp中保存了最近一次的split结果,若这个结果还在被使用中,
就不能用同一个rlrTemp对像来调用,会修改正在被使用中的buf,
procReadData中就是,在第一次split时可能中途还要再split一次其他buf,
这个时候就不能直接传入现有的rlttemp来调用split,必须改用其他一个rlttemp,
不然正在被使用的结果会因为第二次的split而被改写
*/
func (self *MsgQueue) bytesSplit(s []byte, rltTemp *[][]byte) [][]byte {
	frags := BytesSplit(s, msg_delim, *rltTemp)
	// Split内部会扩展self.dataSplitRltTemp底层空间，这里将这个扩展表现到self.dataSplitRltTemp的len字段中，这样之后的Split内部就不调用append了
	if len(frags) > len(*rltTemp) {
		*rltTemp = frags
	}
	return frags
}

func (self *MsgQueue) mergeFrag(frag []byte) {
	self.pushDataTempBuf(frag)
	subFrags := self.bytesSplit(self.dataTempBuf, &self.dataSplitRltTemp2)
	subFragsLen := len(subFrags)

	finalFragOK := false
	for i := 0; i < subFragsLen; i++ {
		/* if frag in 0~subFragsLen-2 invalid, just drop;
		normally they should be valid, so some errors happen, just drop error frag*/
		msg, marks, valid := self.unPackMsg(subFrags[i])
		if valid {
			self.procMsg(marks, msg)
		}
		if i == subFragsLen-1 {
			finalFragOK = valid
		} else {
			if !valid {
				errFrag := subFrags[i]
				errFragLen := len(errFrag)
				if errFragLen > 60 {
					log.Warn(self.name, "err frag:[%v......%v]", errFrag[0:20], errFrag[errFragLen-40:errFragLen])
				} else {
					log.Warn(self.name, "err frag:[%v]", errFrag)
				}
			}
		}
	}
	if finalFragOK {
		self.clearDataTempBuf()
	} else {
		// leave final frag for next merge
		if subFragsLen > 1 {
			self.clearDataTempBuf()
			self.pushDataTempBuf(subFrags[subFragsLen-1])
		}
		// 1 just leave
	}
}
func (self *MsgQueue) procReadData(data []byte) {
	frags := self.bytesSplit(data, &self.dataSplitRltTemp)
	fragCount := len(frags)
	for i := 0; i < fragCount; i++ {
		frag := frags[i]
		msg, marks, valid := self.unPackMsg(frag)
		if valid {
			self.procMsg(marks, msg)
		} else {
			if i == 0 {
				if self.dataTempBufEmpty() {
					self.pushDataTempBuf(frag)
				} else {
					self.mergeFrag(frag)
				}
			} else if i < fragCount-1 {
				log.Warn(self.name, "proc fail mid(0<i<tailIdx) frag:[%v]", frag)
			} else {
				if !self.dataTempBufEmpty() {
					bufLen := len(self.dataTempBuf)
					if bufLen > 60 {
						log.Warn(self.name, "err frag2:[%v......%v]", self.dataTempBuf[0:20], self.dataTempBuf[bufLen-40:bufLen])
					} else {
						log.Warn(self.name, "err frag2:[%v]", self.dataTempBuf)
					}
					self.clearDataTempBuf()
				}
				self.pushDataTempBuf(frag)
			}
		}
	}
}
func (self *MsgQueue) unPackMsg(frag []byte) (msg []byte, marks []uint32, valid bool) {
	fragLen := len(frag)
	if fragLen <= pkg_tail_len {
		return
	}

	l0 := fragLen - len_msg_delim
	if !bytes.Equal(frag[l0:], msg_delim) {
		return
	}

	l1 := l0 - msg_len_len
	msgAndMarksLen := int(binary.BigEndian.Uint32(frag[l1:l0]))
	if fragLen-pkg_tail_len != msgAndMarksLen {
		return
	}

	l2 := l1 - first_byte_len
	l3 := l2 - marks_count_len
	marksCount := int(frag[l3])

	var firstByte byte
	if msgAndMarksLen-marksCount*4 > 0 {
		firstByte = frag[0]
		if frag[l2] != firstByte {
			return
		}
	}

	l4 := l3 - check_code_len
	checkCode := binary.BigEndian.Uint32(frag[l4:l3])
	if checkCode != self.calCheckCode(byte(marksCount), firstByte, msgAndMarksLen) {
		return
	}

	if marksCount > 0 {
		marks = make([]uint32, marksCount)
		le := l4
		for i := marksCount - 1; i >= 0; i-- {
			lb := le - 4
			marks[i] = binary.BigEndian.Uint32(frag[lb:le])
			le = lb
		}
	}

	msg = frag[:msgAndMarksLen-marksCount*4]
	valid = true
	return
}
func (self *MsgQueue) procMsg(marks []uint32, msg []byte) {
	self.procGoLimiter.GetOne()

	if len(msg) > 0 {
		msgBuf := GlobalBufPull(len(msg))
		copy(msgBuf.Value, msg)

		go func() {
			defer self.procGoLimiter.ReturnOne()
			defer GlobalBufPush(msgBuf)

			self.customerMsgProc(marks, msgBuf.Value)

			self.readCount.Add(1)
			self.readBytes.Add(int64(len(msg) + len(marks)*4))
		}()
	} else {
		go func() {
			defer self.procGoLimiter.ReturnOne()

			self.customerMsgProc(marks, nil)

			self.readCount.Add(1)
			self.readBytes.Add(int64(len(marks) * 4))
		}()
	}
}

func (self *MsgQueue) GetRWWaitCnt() (rWaitCnt int, wWaitCnt int) {
	return self.sw.GetWaitProcCount(), len(self.sendMsgChan)
}

// 该函数内部不能加入繁重操作,不能有锁相关操作,且不能开放包含外部函数,因为外部函数可能调用其它含changeStatus的函数,造成死锁
// 这里主要是用于保证changeSign的操作与状态改变一致
func (self *MsgQueue) changeStatus(from int32, to int32, onOk func(), onFail func()) bool {
	self.changeStatusMutex.Lock()
	defer self.changeStatusMutex.Unlock()

	rlt := self.status.CompareAndSwap(from, to)
	if rlt {
		if onOk != nil {
			onOk()
		}
	} else {
		if onFail != nil {
			onFail()
		}
	}
	return rlt
}
